package main

import (
	"database/sql"
	"time"
	"strconv"
	"sync"
	"fmt"
)

func merge() {
	db, _ := sql.Open("mysql", "root:@tcp(localhost:3306)/quotation?charset=utf8")
	db.SetMaxOpenConns(16)
	db.SetMaxIdleConns(16)
	db.SetConnMaxLifetime(time.Second * 3600)

	var channel = make(chan string, 1000000)

	const insert_sql_prefix = "insert into `pair_kline_minute` (platform_id,tpair_id,base_currency," +
		"quote_currency,merge1,kline_start_time,kline_close_time,open_price,close_price,high_price," +
		"low_price,trans_volume,number_of_trades,create_time) values "
	const duplicat_update = " on duplicate key update open_price = IF(unix_timestamp(update_time)>0," +
		"values(open_price),open_price),close_price=IF(unix_timestamp(update_time)>0,values(close_price),close_price)," +
		"high_price=IF(unix_timestamp(update_time)>0,values(high_price),high_price)," +
		"low_price=IF(unix_timestamp(update_time)>0,values(low_price),low_price)," +
		"trans_volume=IF(unix_timestamp(update_time)>0,values(trans_volume),trans_volume)," +
		"number_of_trades=IF(unix_timestamp(update_time)>0,values(number_of_trades),number_of_trades)"
	const row_count = 735441

	start_time := time.Now().Unix()

	//sql生产者
	go func() {
		for i := 0; i <= row_count; i++ {
			var base_currency = "TEST_BASE" + strconv.Itoa(i)
			var quote_currency = "TEST_QUTO" + strconv.Itoa(i)
			var merge1 = base_currency + quote_currency
			var values = "(1,2,'" + base_currency + "','" + quote_currency + "','" + merge1 + "',11111,22222,1,2,3,4,5,6,'2018-06-08'),"
			//交给合并线程处理
			channel <- values
		}
	}()

	//sql消费者
	var get_conter = 0
	var buffer string
	var mege_buffer string
	var wg sync.WaitGroup

	var do_merge = func() {
		if len(buffer) > 0 {
			buffer = buffer[0 : len(buffer)-1]
			mege_buffer = insert_sql_prefix + buffer + duplicat_update
			//panic(mege_buffer)
			buffer = ""
			//异步执行
			wg.Add(1)
			go func() {
				defer wg.Done()
				//开个事务批量提交
				tx, err := db.Begin()
				if nil != err {
					fmt.Println(err)
					return
				}
				_, err = tx.Exec(mege_buffer)
				if nil != err {
					fmt.Println(err)
					tx.Rollback()
					return
				}
				tx.Commit()
			}()
		}
	}

	var timeout = time.NewTicker(500 * time.Millisecond)
	defer timeout.Stop()

	for {
		if get_conter >= row_count {
			//取完了执行最后一段sql，等待执行完成退出
			do_merge()
			wg.Wait()
			break
		}
		select {
		case values := <-channel:
			buffer += values
			get_conter++
		case <-timeout.C:
			do_merge()
		}
	}

	fmt.Println(time.Now().Unix() - start_time)

}
